package data_structure

import (
	"github.com/pkg/errors"
	"math"
	"sync"
	"sync/atomic"
)

type BlockingQueue struct {
	head     *Node
	tail     *Node
	putLock  sync.Mutex
	takeLock sync.Mutex
	// 当队列空的时候，poll协程阻塞在这个条件变量中
	null sync.Cond
	// 当队列满的时候，put协程阻塞在这个条件变量中
	full        sync.Cond
	cap         int32
	currentSize int32
}

const (
	MagicNum int32 = 0
)

// 可选参数
type QueueOption func(queue *BlockingQueue)

func WithCap(cap int32) QueueOption {
	return func(queue *BlockingQueue) {
		queue.cap = cap
	}
}

func NewBlockingQueue(option ...QueueOption) *BlockingQueue {
	// 先初始化一个虚节点
	dummy := NewNode(MagicNum, nil)
	putLock := sync.Mutex{}
	takeLock := sync.Mutex{}
	full := sync.NewCond(&putLock)
	null := sync.NewCond(&takeLock)
	queue := &BlockingQueue{
		head:        dummy,
		tail:        dummy,
		putLock:     putLock,
		takeLock:    takeLock,
		full:        *full,
		null:        *null,
		currentSize: 0,
		cap:         0,
	}
	for _, queueOption := range option {
		queueOption(queue)
	}
	if queue.cap == 0 {
		queue.cap = math.MaxInt32
	}
	return queue
}

// 协程非安全入队
func (q *BlockingQueue) enqueue(newNode *Node) {
	q.tail.next = newNode
	q.tail = q.tail.next
}

// 协程非安全出队
func (q *BlockingQueue) dequeue() interface{} {
	h := q.head
	first := h.next
	if first == nil {
		return nil
	}
	// 帮助gc
	h.next = h
	v := first.val
	q.head = first
	q.head.val = MagicNum
	return v
}

// 协程安全入队
func (q *BlockingQueue) Push(v interface{}) error {
	if v == nil {
		return errors.New("the integer element is nil")
	}
	newNode := NewNode(v, nil)
	q.putLock.Lock()
	defer q.putLock.Unlock()
	// 当队列满的时候，一直在条件变量中等待。
	// 必须使用while循环。当前协程被唤醒后，
	// 可能第一次没抢到锁，其他put协程将元素
	// 加入到队列中(而且队列满了)。 然后第二次，
	// 当前协程抢到锁之后，如果不用while循环而用if，
	// 那么就会走后续流程，但实际上现在队列已满，不符合预期。
	// 因此要用while循环一直循环判断，防止以上的情况的出现。
	for q.cap == q.currentSize {
		q.full.Wait()
	}
	q.enqueue(newNode)
	// cas增加容量
	newCap := atomic.AddInt32(&q.currentSize, 1)
	// 如果当前队列容量没满，那么就应该唤醒阻塞在full变量的put协程，通知它们可以继续往队列注入元素。
	if newCap < q.cap {
		q.full.Signal()
	}
	// 如果队列从空变成了有元素，那么就通知那些阻塞在null变量的poll协程，通知它们可以从队列索取元素。
	if newCap == 1 {
		q.notifyNull()
	}
	return nil
}

// 协程安全出队
func (q *BlockingQueue) Poll() interface{} {
	q.takeLock.Lock()
	defer q.takeLock.Unlock()
	for q.cap == 0 {
		q.null.Wait()
	}
	res := q.dequeue()
	// cas减少容量
	newCap := atomic.AddInt32(&q.currentSize, -1)
	// 如果当前队列容量没空，那么就应该唤醒阻塞在null变量的poll协程，通知它们可以继续从队列索取元素。
	if newCap > 0 {
		q.null.Signal()
	}
	// 如果队列从满变成了不满，那么就通知那些阻塞在full变量的put协程，通知它们可以从队列注入元素。
	if newCap == newCap-1 {
		q.notifyFull()
	}
	return res
}

func (q *BlockingQueue) Offer(v interface{}) {
	newNode := NewNode(v, nil)
	q.enqueue(newNode)
}

func (q *BlockingQueue) Pull() interface{} {
	return q.dequeue()
}

func (q *BlockingQueue) IsEmpty() bool {
	return q.currentSize == 0
}

// 获取队列现在的长度
func (q *BlockingQueue) Len() int32 {
	return q.currentSize
}

// put协程唤醒阻塞在null变量的协程
func (q *BlockingQueue) notifyNull() {
	q.takeLock.Lock()
	defer q.takeLock.Unlock()
	q.null.Signal()
}

// poll协程唤醒阻塞在full变量的协程
func (q *BlockingQueue) notifyFull() {
	q.putLock.Lock()
	defer q.putLock.Unlock()
	q.full.Signal()
}
